package com.we.risk.phoneRecord.stage2.dimuser;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.risk.phoneRecord.common.Stage2GlobalSetting;
import com.weicai.caesar.CaesarUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Deprecated
public class TableUserPidBroadCast {
    public static final String KFKTABLE = "fdm.hhl01_user_pid";
    public static final String QUERYSQL =
            "select uid, pid, updated_at "
                    + " from "
                    + KFKTABLE
                    + " where deleted_at is null "
                    + "and updated_at is not null "
                    + "and pid is not null";
    public static final int PIDLENGTH = 6;
    public static final String ENCRYPTPID = "pid";
    public static final String INPUT_KEY_BY = "uid";
    public static final String PID_AREA_BROADCAST_DESC = "pid_area_broadcast_desc";
    public static final int PID_AREA_BROADCAST_WAIT_SEC = 5000;
    public static final String SPLITWORD = "#";

    public long uid;
    public String pid;
    public int updated_at;

    public TableUserPidBroadCast() {}

    public TableUserPidBroadCast(long uid, String pid, int updated_at) {
        this.uid = uid;
        this.pid = pid;
        this.updated_at = updated_at;
    }

    public static Logger LOG = LoggerFactory.getLogger(TableUserPidBroadCast.class);

    public static TableUserPidBroadCast init(long uid) {
        return new TableUserPidBroadCast(uid, "", -1);
    }

    public static SingleOutputStreamOperator<String> getUserPidProcessedStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableUserPidBroadCast.QUERYSQL);
        DataStream<Tuple2<Boolean, TableUserPidBroadCast>> inputStream =
                tableEnv.toRetractStream(tableRes, TableUserPidBroadCast.class);

        KafkaSource<String> pidAreaSource =
                KafkaSource.<String>builder()
                        .setBootstrapServers(Stage2GlobalSetting.MOBILE_AREA_KFK_BOOTADD)
                        .setTopics(Stage2GlobalSetting.PID_AREA_KFK_TOPIC)
                        .setGroupId(Stage2GlobalSetting.PID_AREA_KFK_GRP_ID)
                        .setStartingOffsets(OffsetsInitializer.earliest())
                        .setValueOnlyDeserializer(new SimpleStringSchema())
                        .build();

        SingleOutputStreamOperator<String> pidAreaInput =
                env.fromSource(pidAreaSource, WatermarkStrategy.noWatermarks(), "idArea")
                        .uid("pidArea_input");
        MapStateDescriptor<String, String> pidAreaBroadcastDesc =
                new MapStateDescriptor<>(PID_AREA_BROADCAST_DESC, String.class, String.class);
        BroadcastStream<String> pidAreaBroadcastStream =
                pidAreaInput.broadcast(pidAreaBroadcastDesc);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableUserPidBroadCast>>() {
                                    @Override
                                    public boolean filter(
                                            Tuple2<Boolean, TableUserPidBroadCast> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TableUserPid-filter")
                        .keyBy(new TableUserPidKeySelectorFunc())
                        .flatMap(new TableUserPidFlatMapFunc())
                        .keyBy(
                                new KeySelector<String, Long>() {
                                    @Override
                                    public Long getKey(String value) throws Exception {
                                        return JSONObject.parseObject(value).getLong(INPUT_KEY_BY);
                                    }
                                })
                        .connect(pidAreaBroadcastStream)
                        .process(new UserJoinPidAreaProcessFunc());

        return out;
    }

    private static class TableUserPidKeySelectorFunc
            implements KeySelector<Tuple2<Boolean, TableUserPidBroadCast>, Long> {
        @Override
        public Long getKey(Tuple2<Boolean, TableUserPidBroadCast> value) throws Exception {
            return value.f1.getUid();
        }
    }

    private static class TableUserPidFlatMapFunc
            extends RichFlatMapFunction<Tuple2<Boolean, TableUserPidBroadCast>, String> {
        ValueState<TableUserPidBroadCast> tableUserPidVSatate;

        @Override
        public void open(Configuration parameters) throws Exception {
            tableUserPidVSatate =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<TableUserPidBroadCast>(
                                            "TableUserPidVSatate", TableUserPidBroadCast.class));
        }

        @Override
        public void flatMap(Tuple2<Boolean, TableUserPidBroadCast> value, Collector<String> out)
                throws Exception {
            try {
                long uid = value.f1.getUid();
                int updated_at = value.f1.getUpdated_at();
                String pid = value.f1.getPid();
                if (CaesarUtil.isEncrypted(pid)) {
                    pid = CaesarUtil.decode(pid);
                }
                if (pid.length() >= PIDLENGTH) {
                    pid = pid.substring(0, PIDLENGTH);
                }

                TableUserPidBroadCast tmpState = tableUserPidVSatate.value();
                TableUserPidBroadCast curSTate =
                        tmpState == null ? TableUserPidBroadCast.init(uid) : tmpState;

                if (updated_at > curSTate.getUpdated_at()) {
                    if (!pid.equals(curSTate.pid)) {
                        JSONObject json = new JSONObject();
                        json.put("uid", uid);
                        json.put("pid", pid);

                        out.collect(json.toString());

                        curSTate.updated_at = updated_at;
                        curSTate.pid = pid;
                        tableUserPidVSatate.update(curSTate);
                    }
                }
            } catch (Exception e) {
                LOG.error("TableUserPid flatMap Exception: " + e.toString());
                LOG.error("TableUserPid flatMap Exception input: " + value);
                e.printStackTrace();
            }
        }
    }

    private static class UserJoinPidAreaProcessFunc
            extends KeyedBroadcastProcessFunction<String, String, String, String> {
        MapState<String, String> pidAreaWaitListMState;
        public static final MapStateDescriptor<String, String> pidAreaBrodcstdesc =
                new MapStateDescriptor<>(PID_AREA_BROADCAST_DESC, String.class, String.class);
        long startTime = 0L;
        //        Long bStateSize = 0L;
        @Override
        public void open(Configuration parameters) throws Exception {
            pidAreaWaitListMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<String, String>(
                                            "PidAreaWaitListMState", String.class, String.class));
            startTime = System.currentTimeMillis();
        }

        @Override
        public void processElement(String value, ReadOnlyContext ctx, Collector<String> out)
                throws Exception {
            try {
                long timediff = (ctx.timerService().currentProcessingTime() - startTime);
                ReadOnlyBroadcastState<String, String> readBState =
                        ctx.getBroadcastState(pidAreaBrodcstdesc);
                if (timediff >= PID_AREA_BROADCAST_WAIT_SEC) {
                    // real process element
                    if (readBState != null) {
                        if (!pidAreaWaitListMState.isEmpty()) {
                            pidAreaWaitListMState.clear();
                        }
                        ProcessData(value, readBState, out);
                    }
                } else {
                    LOG.debug(
                            "UserJoinPidAreaProcessFunc processElement addinto waitList value: "
                                    + value);
                    pidAreaWaitListMState.put(INPUT_KEY_BY, value);
                    long now = ctx.timerService().currentProcessingTime();
                    ctx.timerService()
                            .registerProcessingTimeTimer(now + PID_AREA_BROADCAST_WAIT_SEC);
                }

            } catch (Exception e) {
                LOG.error("UserJoinPidAreaProcessFunc processElement exception: " + e.toString());
                LOG.error("UserJoinPidAreaProcessFunc processElement error input: " + value);
                e.printStackTrace();
            }
        }

        /** {"id":5373,"id_no":"654001","province":"新疆维吾尔自治区","city":"伊犁哈萨克自治州","area":"奎屯市"} */
        @Override
        public void processBroadcastElement(String value, Context ctx, Collector<String> out)
                throws Exception {
            try {
                BroadcastState<String, String> writeBState =
                        ctx.getBroadcastState(pidAreaBrodcstdesc);
                if (value != null) {
                    JSONObject inputJson = JSONObject.parseObject(value);
                    String id_no = inputJson.getString("id_no");
                    String province = inputJson.getString("province");
                    String city = inputJson.getString("city");
                    String area = inputJson.getString("area");

                    /** 维表异常数据处理 */
                    if (province == null || province.equals("")) {
                        province = "null";
                    }

                    if (city == null || city.equals("")) {
                        city = "null";
                    }

                    if (area == null || area.equals("")) {
                        area = "null";
                    }

                    writeBState.put(id_no, province + SPLITWORD + city + SPLITWORD + area);
                    //                    bStateSize += 1;
                }
            } catch (Exception e) {
                LOG.error(
                        "UserJoinPidAreaProcessFunc processBroadcastElement exception: "
                                + e.toString());
                LOG.error(
                        "UserJoinPidAreaProcessFunc processBroadcastElement error input: " + value);
                e.printStackTrace();
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out)
                throws Exception {
            super.onTimer(timestamp, ctx, out);
            try {
                LOG.debug(
                        "debug UserJoinPidAreaProcessFunc onTimer triggered isEmpty: "
                                + pidAreaWaitListMState.isEmpty());
                if (!pidAreaWaitListMState.isEmpty()) {
                    String waitMsg = pidAreaWaitListMState.get(INPUT_KEY_BY);
                    ReadOnlyBroadcastState<String, String> readBState =
                            ctx.getBroadcastState(pidAreaBrodcstdesc);
                    ProcessData(waitMsg, readBState, out);
                    pidAreaWaitListMState.clear();
                }
                ctx.timerService().deleteProcessingTimeTimer(timestamp);
            } catch (Exception e) {
                LOG.error("UserJoinPidAreaProcessFunc onTimer exception: " + e.toString());
                e.printStackTrace();
            }
        }

        private void ProcessData(
                String value,
                ReadOnlyBroadcastState<String, String> readBState,
                Collector<String> out)
                throws Exception {
            JSONObject inputJson = JSONObject.parseObject(value);
            String pid = inputJson.getString(ENCRYPTPID);
            String res = readBState.get(pid);
            if (res != null) {
                String[] splits = res.split(SPLITWORD);
                if (splits.length == 3) {
                    inputJson.put("pid_prov", splits[0].equals("null") ? "" : splits[0]);
                    inputJson.put("pid_city", splits[1].equals("null") ? "" : splits[1]);
                    inputJson.put("pid_area", splits[2].equals("null") ? "" : splits[2]);
                    //                    LOG.debug("Debug UserJoinPidAreaProcessFunc ProcessData
                    // bStateSize:" + bStateSize + " res: " +
                    // inputJson.toString(SerializerFeature.WriteMapNullValue));
                    out.collect(inputJson.toString());
                } else {
                    LOG.error("UserJoinPidAreaProcessFunc ProcessData error pid: " + pid);
                }
            } else {
                /** 未关联到维表数据不向下输出 */
                LOG.warn("UserJPidFunc ProcData pid: " + pid);
            }
        }
    }

    public long getUid() {
        return uid;
    }

    public int getUpdated_at() {
        return updated_at;
    }

    public String getPid() {
        return pid;
    }
}
